package com.cloudcanal.processor;

import java.util.List;

import javax.sql.DataSource;

import com.clougence.cloudcanal.sdk.api.CloudCanalProcessorV2;
import com.clougence.cloudcanal.sdk.api.JavaDsType;
import com.clougence.cloudcanal.sdk.api.ProcessorContext;
import com.clougence.cloudcanal.sdk.api.modelv2.CustomData;
import com.clougence.cloudcanal.sdk.api.modelv2.SchemaInfo;

public class DataChangeProcessor implements CloudCanalProcessorV2{
	
	private DataSource srcDataSource;
	private DataSource dstDataSource;
	@Override
	public void start(ProcessorContext context) {
		if (context.getSrcDsType().getContextDsType() != JavaDsType.JdbcDataSource) {
            throw new IllegalArgumentException("src ds type is not JdbcDataSource");
        }
		if (context.getDstDsType().getContextDsType() != JavaDsType.JdbcDataSource) {
            throw new IllegalArgumentException("det ds type is not JdbcDataSource");
        }
		srcDataSource = (DataSource) context.getSrcRdbDs();
		dstDataSource = (DataSource) context.getDstRdbDs();
	}

	@Override
	public List<CustomData> process(CustomData data) {
		SchemaInfo schemaInfo = data.getSchemaInfo();
		RecordModifyConvert convert = RecordModifyConvertFactory.get(schemaInfo);
		return convert.modify(data, srcDataSource, dstDataSource);
	}

	@Override
	public void stop() {
		
	}
}
